package com.lagou.hbase.processsor;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class DeleteProcessor extends BaseRegionObserver {

    /*
       重写postDelete方法，监听到一方好友删除时，另一方相应地删除该好友
     */
    @Override
    public void postDelete(ObserverContext<RegionCoprocessorEnvironment> e, Delete delete, WALEdit edit, Durability durability) throws IOException {

        //获取table对象
        HTable relations = (HTable) e.getEnvironment().getTable(TableName.valueOf("relations"));

        List deletes = new ArrayList<Delete>();
        Scan scan = new Scan();

        //取得所有cell
        List<Cell> cells = delete.getFamilyCellMap().get(Bytes.toBytes("friends"));
        for(Cell cell : cells){
            //取得删除者
            byte[] row = CellUtil.cloneRow(cell);
            //取得被删除的列
            byte[] delCol = CellUtil.cloneQualifier(cell);
            //查看被删者是否也有删除者好友
            scan.setStartRow(delCol);
            ResultScanner resultScanner = relations.getScanner(scan);
            Result result = resultScanner.next();
            //若有，被删者也同时删除好友
            if(result != null && !result.isEmpty()){
                Cell[] cells1 = result.rawCells();
                for(Cell cell1 : cells1){
                    if(new String(row).equals(new String(CellUtil.cloneQualifier(cell1)))){
                        Delete delete1 = new Delete(delCol);
                        delete1.addColumn(CellUtil.cloneFamily(cell1),row);
                        deletes.add(delete1);
                    }
                }
            }
        }
        //执行批量删除
        relations.delete(deletes);
        //关闭table
        relations.close();
    }
}
